package com.surfilter.massdata.es.core.client;

import java.net.InetAddress;
import java.net.UnknownHostException;

import org.apache.log4j.Logger;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
import org.elasticsearch.action.admin.indices.forcemerge.ForceMergeRequest;
import org.elasticsearch.action.admin.indices.forcemerge.ForceMergeResponse;
import org.elasticsearch.action.admin.indices.mapping.put.PutMappingRequest;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.Requests;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.metadata.MetaData;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.transport.TransportAddress;
import org.elasticsearch.plugin.deletebyquery.DeleteByQueryPlugin;

/**
 * 获取ES Client单例
 * 
 * @ClassName: ClusterClient
 * @Description: 
 * @author liuchen
 * @date 2013-4-25 上午10:20:35
 *
 */
public class EsClient {
	protected final Logger logger = Logger.getLogger(getClass());
	
	
	private EsEnv esEnv;
	//ES Client
	private Client client;
	
	protected EsClient(String clusterName, int numberOfShards,int numberOfReplicas) {
		Settings settings = Settings.settingsBuilder()
				.put("cluster.name", clusterName)
				.put("number_of_shards", numberOfShards)
				.put("number_of_replicas", numberOfReplicas).build();
		client = TransportClient.builder().settings(settings).build();
	}
	
	protected EsClient addTransport(String host, int port) {
		try {
			((TransportClient) client).addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(host), port));
		} catch (UnknownHostException e) {
			logger.error("add es transportAddress error host:" + host, e);
		}
		return this;
	}
	
	
	private void buildClient(){
		Settings settings = Settings.settingsBuilder()
				.put("cluster.name", esEnv.getClusterName())
				.put("transport.tcp.compress", true)
				.build();
		String[] arrIp = esEnv.getIp().split(",");
		String[] arrPort = esEnv.getPort().split(",");
		TransportAddress[] addressArr = new TransportAddress[arrPort.length];
		for (int i = 0 ,size = arrIp.length; i < size; i++) {
			String objIp = arrIp[i];
			int port = 9300;
			try {
				port = Integer.valueOf(arrPort[i]);
				addressArr[i] = new InetSocketTransportAddress(InetAddress.getByName(objIp), port);
			} catch (Exception e) {
				logger.error(" host or port trans error !");
			}
		}
		client = TransportClient.builder().settings(settings)
				//TODO 添加删除插件
				.addPlugin(DeleteByQueryPlugin.class)
				.build().addTransportAddresses(addressArr);
		logger.info("开辟集群连接,address:"+"\t连接对象"+client);
	}
	
	public Client getClient() {
		return client;
	}

	public void rebuildClient(){
		logger.info("上次client连接发生错误,重新开辟连接!");
		if(client != null){
			close();
		}
		buildClient();
	}
	
	public EsClient(EsEnv esEnv) {
		this.esEnv = esEnv;
		buildClient();
	}
	
	public BulkRequestBuilder getBulkRequestBuilder(){
		return client.prepareBulk();
	}
	
	public IndexRequestBuilder getIndexRequestBuilder(){
		return client.prepareIndex();
	}
	
	public SearchRequestBuilder getEsSearch(){
		return client.prepareSearch();
	}
	
	public void esRefresh(){
		client.admin().indices().prepareRefresh().execute().actionGet();
	}
	
	/**
	 * 获取ES服务器的所有打开的索引
	 * 
	 * @Description: 
	 * @return
	 */
	public ClusterHealthResponse getClusterHealthResponse(){
		return client.admin().cluster().health(Requests.clusterHealthRequest().waitForGreenStatus()).actionGet();
	}
	
	/**
	 * 获取ES服务器的所有索引（包括打开和关闭的索引）
	 * 
	 * @Description: 
	 * @return
	 */
	public MetaData getMetaData(){
		ClusterState state = client.admin().cluster().prepareState().execute().actionGet().getState();
		return state.getMetaData();
	}
	
	
	public boolean isExists(String indexName){
		return client.admin().indices().prepareExists(indexName).execute().actionGet().isExists();
	}
	
	public boolean closeIndex(String indexName){
		return client.admin().indices().prepareClose(indexName).execute().actionGet().isAcknowledged();
	}
	
	public boolean deleteIndex(String indexName){
		return client.admin().indices().prepareDelete(indexName).execute().actionGet().isAcknowledged();
	}
	
	public boolean createIndex(Settings settings, String indexName){
		return client.admin().indices().prepareCreate(indexName).setSettings(settings).execute().actionGet().isAcknowledged();
	}
	
	public void createAlias(String indexName, String aliasName){
		client.admin().indices().prepareAliases().addAlias(indexName, aliasName).execute().actionGet();
	}
	
	public boolean putMapping(PutMappingRequest mappingRequest){
		return client.admin().indices().putMapping(mappingRequest).actionGet().isAcknowledged();
	}
	
	public ForceMergeResponse optimize(String indexName) {
		return client.admin().indices()
				.forceMerge(new ForceMergeRequest(indexName).flush(true)// 是否合并完成后进行flush操作，默认为true
						.maxNumSegments(1)// 最大合并segment数
						.onlyExpungeDeletes(true)// 是否只删除已做删除标记的记录
		).actionGet();
	}	
/*	public OptimizeResponse optimize(String indexName){
		
		OptimizeRequest optRequest = new OptimizeRequest(new String[] { indexName });
		// optRequest.flush(true);
		optRequest.maxNumSegments(1);
		OptimizeResponse optResponse = client.admin().indices().optimize(
					optRequest).actionGet();
		return optResponse;
	}*/
	
	
	
	/**
	 * 关闭ES客户端
	 * 
	 * @Description:
	 */
	public void close(){
		logger.info("\t关闭连接对象"+client);
		client.close();
	}
	
	
}
